package com.atguigu.flink.tableapi;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.OverWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by Smexy on 2023/3/5
 */
public class Demo10_OverWindow
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, ts) -> e.getTs());

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            //在流中产生水印
            .assignTimestampsAndWatermarks(watermarkStrategy);

        Table table = tableEnv.fromDataStream(ds, $("id"),$("ts"),$("vc"),
            $("pt").proctime(),
            $("et").rowtime());

        /*
            select xxx, 开窗函数() over( 定义窗口 )
                开窗函数:
                        在hive中：
                            max,min,sum,count,avg
                            排名函数: rank,row_number..
                            窗口函数: lead,lag,first_value,last_value

                定义窗口:  partitioned by xxx order by xxx rows|range between 上界  and  下界


               flink中:
                    rows: 以行数定义窗口
                    range: 以时间定义窗口
                    order by xxx: xxx只能写时间字段，不能写别的.
                                    xxx只能写一个字段，如果是stream，可以是et也可以是pt
                                    是et的话，只有当前水印达到事件时间，才会触发运算。
                    下界：不能超过流的当前数据，最多只能是到当前行或时间范围！

               通过Over对象定义！
                    Over窗口和 Window窗口(flink中的窗口，分为事件时间，处理时间)不一样

                基于rows运算，PT方便，et会延迟一条数据。
         */
        //基于rows   上无边界 到  当前行
        OverWindow w1 = Over.partitionBy($("id")).orderBy($("et")).preceding(UNBOUNDED_ROW).following(CURRENT_ROW).as("w");
        //基于rows   前2行 到  当前行
        OverWindow w2 = Over.partitionBy($("id")).orderBy($("pt")).preceding(rowInterval(2l)).following(CURRENT_ROW).as("w");
        //不能超过当前范围，错误示范
        OverWindow w3 = Over.partitionBy($("id")).orderBy($("et")).preceding(UNBOUNDED_ROW).following(rowInterval(2l)).as("w");


        //基于range  上无边界 到  当前时间范围
        OverWindow w4 = Over.partitionBy($("id")).orderBy($("et")).preceding(UNBOUNDED_RANGE).following(CURRENT_RANGE).as("w");
        //基于range  2s前 到  当前时间范围
        OverWindow w5 = Over.partitionBy($("id")).orderBy($("et")).preceding(lit(2).seconds()).following(CURRENT_RANGE).as("w");
        // 错误示范，超过了当前时间范围
        //OverWindow w6 = Over.partitionBy($("id")).orderBy($("et")).preceding(lit(2).seconds()).following(lit(2).second()).as("w");

        table
            .window(w5)
            .select($("id"),$("et"),
                    $("vc").sum().over($("w")).as("sumVC"))
            .execute()
            .print();

    }
}
